package com.u.statistics.job;

import com.alibaba.fastjson.JSON;
import com.u.statistics.model.UserSleepStatus;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
import java.util.Properties;
import java.util.Random;

/**
 * @program: u-InternetOfThings
 * @description: sleep状态流
 * @author: Alex Wu
 * @createDate: 2025-03-20 17:54
 **/

@Slf4j
public class SleepStatusStream {

    private static final String ENV_KAFKA_SERVER = "KAFKA_SERVER";
    private static final String KAFKA_BEHAVIOR_LOG_TOPIC = "test";
    private static final String KAFKA_BEHAVIOR_LOG_GROUP_ID = "test_group";
    private static final String JOB_NAME_BEHAVIOR_LOG = "Flink_Behavior_Log";

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 配置 Kafka 数据源（假设从 Kafka 获取用户状态数据）
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.102.8:9092");
        properties.setProperty("group.id", "flink-group");

        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
                "test",
                new SimpleStringSchema(),
                properties
        );

        // 读取 Kafka 数据流
        DataStreamSource<String> sourceStream = env.addSource(new SourceFunction<String>() {
            private boolean flag = true;

            private Integer startTime = 1000;

            @Override
            public void run(SourceContext<String> ctx) throws Exception {

                System.out.println("开始生成数据");
                Random random = new Random();
                while (flag) {
                    Long userId = 1L;
                    String status = "sleep";
                    startTime += 1000;
//                    Long eventTime = Instant.parse("2024-11-15T00:00:00Z").toEpochMilli() - random.nextInt(5) * 1000 + startTime;
                    Long eventTime = System.currentTimeMillis() - random.nextInt(5) * 1000;
                    Integer breathingRate =  random.nextInt(100) ;
                    Integer heartRate =  random.nextInt(100) ;
                    Integer turnsNum =  random.nextInt(100) ;
                    UserSleepStatus userSleepStatus = new UserSleepStatus(userId, status, eventTime, breathingRate, heartRate, turnsNum);



                    ctx.collect(JSON.toJSONString(userSleepStatus));
//                    ctx.collect(userSleepStatus);


                    long currentTimeMillis = System.currentTimeMillis();

                    // 创建SimpleDateFormat对象，指定格式
                    SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");

                    // 将时间戳转换为Date对象
                    Date date = new Date(currentTimeMillis);

                    // 格式化Date对象，得到字符串形式的时间
                    String formattedDate = sdf.format(date);

                    // 打印格式化后的时间
                    log.info("Formatted time: " + formattedDate + " event time: " + sdf.format(userSleepStatus.getEventTime()) + " Order info: " + userSleepStatus );



                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {
                flag = false;
            }
        });

//        DataStream<String> sourceStream = env.addSource(kafkaConsumer);

// 1. 绑定水位线 告警
        SingleOutputStreamOperator<UserSleepStatus> userSleepStatusWatermark = sourceStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<String>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((event, timestamp) -> {
                                    UserSleepStatus userSleepStatus = JSON.parseObject(event, UserSleepStatus.class);
                                    SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
                                    System.out.println("水位线分配: event time = " + sdf.format(userSleepStatus.getEventTime()));
                                    return userSleepStatus.getEventTime();
                                })
                )
                .map(new MapFunction<String, UserSleepStatus>() {
                    @Override
                    public UserSleepStatus map(String userSleepStatusStr) {
                        UserSleepStatus userSleepStatus = JSON.parseObject(userSleepStatusStr, UserSleepStatus.class);
                        SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");

                        String eventFormatted = sdf.format(userSleepStatus.getEventTime());
                        System.out.println("Formatted event time: " + eventFormatted +
                                " --> UserSleepStatus: " + userSleepStatus);
                        return userSleepStatus;
                    }
                });



        SingleOutputStreamOperator<String> process = userSleepStatusWatermark
                .keyBy(UserSleepStatus::getUserId)  // RickProcessFunc 中使用的 Key
                .window(TumblingEventTimeWindows.of(Time.seconds(5))) // 1分钟窗口
                .process(new ProcessWindowFunction<UserSleepStatus, String, Long, TimeWindow>() {



                    @Override
                    public void process(Long key, Context context, Iterable<UserSleepStatus> elements, Collector<String> collector) throws Exception {


                        for (UserSleepStatus userSleepStatus : elements) {
                            if(userSleepStatus.getHeartRate() > 100){
                                log.info("心率过高, heartRate: {}", userSleepStatus.getHeartRate());
                            }
                            log.info("In Windows -->" + userSleepStatus);    // 打印此窗口内的数据
                        }

                        long windowStart = context.window().getStart();
                        long windowEnd = context.window().getEnd();
                        // Format window timestamp
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        String formattedWindowStart = sdf.format(new Date(windowStart));
                        String formattedWindowEnd = sdf.format(new Date(windowEnd));
                        String curTime = sdf.format(new Date());

                        String format = String.format("CurTime: %s, Window [%s - %s], UserId: %d",
                                curTime, formattedWindowStart, formattedWindowEnd, key);
                        log.info(format);

                        collector.collect("============");
                    }
                });


        // 2-保存状态
        userSleepStatusWatermark.keyBy(UserSleepStatus::getUserId)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .process(new ProcessWindowFunction<UserSleepStatus, String, Long, TimeWindow>() {

                    private transient ValueState<String> lastState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        ValueStateDescriptor stateDescriptor = new ValueStateDescriptor("lastState", String.class, "NONE");
                        lastState = getRuntimeContext().getState(stateDescriptor);

                    }

                    @Override
                    public void process(Long key, Context context, Iterable<UserSleepStatus> elements, Collector<String> out) throws Exception {
                        int turnCount = 0;

                        for(UserSleepStatus status : elements){
                            turnCount += status.getTurnsNum();
                        }

                        // TODO 1. 当前面没有状态或者是清醒，最近10分钟内翻身次数>10次，则开始睡眠
                        //2. 前面状态为开始或者睡眠，最近10分钟内翻身次数>5<10次，则是浅睡
                        //3. 前面状态为开始或者睡眠，最近10分钟内翻身次数<5次,则是深睡
                        //4. 前面状态为开始或者睡眠，最近10分钟内翻身次数>10次，则是清醒
                        //5. 其他状态为清醒

                        String currentState = lastState.value();

                        if((lastState.value().equals("NONE") || lastState.value().equals("WAKE")) && turnCount > 10) {
                            currentState = "START";
                        }else if (lastState.value().equals("START") || lastState.value().equals("SLEEP")) {
                            if(turnCount >= 5 && turnCount <= 10){
                                currentState = "LIGHT_SLEEP";
                            }else if (turnCount < 5){
                                currentState = "DEEP_SLEEP";
                            }else if (turnCount > 10){
                                currentState = "WAKE";
                            }
                        }else{
                            currentState = "WAKE";

                        }
                        lastState.update(currentState);

                        // 获取窗口开始和结束时间
                        long windowStart = context.window().getStart();
                        long windowEnd = context.window().getEnd();
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        String formattedWindowStart = sdf.format(new Date(windowStart));
                        String formattedWindowEnd = sdf.format(new Date(windowEnd));
                        String curTime = sdf.format(new Date());

                        // 输出当前窗口的信息
                        String format = String.format("CurTime: %s, Window [%s - %s], UserId: %d, TurnCount: %d, State: %s",
                                curTime, formattedWindowStart, formattedWindowEnd, key, turnCount, currentState);
                        log.info(format);

                        // 输出结果
                        out.collect(format);
                    }
                });

        process.print();



        // 3-控制其他设备
        userSleepStatusWatermark.keyBy(UserSleepStatus::getUserId)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .process(new ProcessWindowFunction<UserSleepStatus, String, Long, TimeWindow>() {

                             @Override
                             public void process(Long key, Context context, Iterable<UserSleepStatus> elements, Collector<String> out) throws Exception {
                                 int turnCount = 0;


                                 for(UserSleepStatus status : elements){
                                     turnCount += status.getTurnsNum();
                                 }
                                // 控制其他设备

                             }

                         });


        // 4-存储HBase 存储到MySQL
//        finalAggregatedStream.addSink(new MySQLSink());*/

        env.execute("User Status Calculation");
    }

    @Data
    @AllArgsConstructor
    static class TimestampedEvent<T> {
        private final long timestamp;
        private final T event;
    }
}

